package com.chencong.online.function;

import com.chencong.online.bean.UniqueVisitorBean;
import com.chencong.online.bean.UserBehaviorBean;
import com.chencong.online.utils.BloomFilterUtil;
import com.chencong.online.utils.RedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.junit.Test;
import redis.clients.jedis.Jedis;

/**
 * @program: user-behavior-analysis-online
 * @ClassName UVBloomFilterProcessFunc
 * @description:利用布隆过滤器进行去重
 * @author: chencong
 * @create: 2021-12-28 16:19
 **/
public class UVBloomFilterProcessFunc extends ProcessAllWindowFunction<UserBehaviorBean, UniqueVisitorBean, TimeWindow> {
    //定义Jedis客户端和布隆过滤器
    Jedis jedis;
    BloomFilterUtil bloomFilterUtil;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = RedisUtil.getJedis();

        bloomFilterUtil = new BloomFilterUtil(1 << 29);// 要处理1亿个数据，用64MB大小的位图
    }

    @Override
    public void process(ProcessAllWindowFunction<UserBehaviorBean, UniqueVisitorBean, TimeWindow>.Context context, Iterable<UserBehaviorBean> elements, Collector<UniqueVisitorBean> out) throws Exception {

        //将位图和窗口count全部存入redis，用windowEnd作为key
        long windowEnd = context.window().getEnd();
        String bitmapKey = String.valueOf(windowEnd);
        //redis怎么存：(key,hash(key,value))
        //把count存成一张hash表
        String countHashName = "uv_count";
        String countKey = String.valueOf(windowEnd);

        //1、取出当前的userId
        Long userId = elements.iterator().next().getUserId();

        //2、计算位图中的offset
        Long offset = bloomFilterUtil.hashCode(userId.toString(), 80);
//        System.out.println("看看这个位图是多少=" + offset);

        //3、用redis的getBit命令，判断对应位置的值
        Boolean isExist = jedis.getbit(bitmapKey, offset);
        System.out.println("对于位置是否有数据=" + isExist);
//        System.out.println("test");
        if (!isExist) {
            //如果不存在，对应位图置1
            jedis.setbit(bitmapKey, offset, true);
            System.out.println("test1!!");
            //更新redis中保存的count值
            Long uvCount = 0L;//初始化count值
            String uvCountString = jedis.hget(countHashName, countKey);
            //如果不等于null和空字符串
            System.out.println("对于位置是否有数据=" + StringUtils.isNotEmpty(uvCountString));
            if (StringUtils.isNotEmpty(uvCountString)) {
                System.out.println("test2!!!");
                uvCount = Long.valueOf(uvCountString);
                //新来一条数据+1
                String uvCountAdd = String.valueOf(uvCount + 1L);
                //最后更新redis中的count
                jedis.hset(countHashName, countKey, uvCountAdd);

                out.collect(new UniqueVisitorBean(String.valueOf(windowEnd), Long.valueOf(uvCountAdd)));
            }
        }
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }

}
